package com.cxdr.awsips.distribute.processor;

import java.sql.CallableStatement;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Types;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.apache.log4j.Logger;

import com.cxdr.awsips.common.db.DBHelper;
import com.cxdr.awsips.common.logger.LogFactory;
import com.cxdr.awsips.common.thread.BaseProcessor;
import com.cxdr.awsips.distribute.bean.DistributionLog;
import com.cxdr.awsips.distribute.config.SystemConfig;
import com.cxdr.awsips.distribute.pool.BaseQueue;

/**
 * 写入分发日志线程
 * @author pugang
 *
 */
public class StoreDistributeProccessor extends BaseProcessor {
	public static final BaseQueue<DistributionLog> DistrbuteFileLogQueue = new BaseQueue<DistributionLog>();
	private static Logger logger = LogFactory.makeNewLogger(StoreDistributeProccessor.class);
	private static final int BATCH_SIZE = Integer.parseInt(SystemConfig.get("num.batchsize"));
	private StoreThread[] ths = new StoreThread[Integer.parseInt(SystemConfig.get("num.reportlog-threads"))];
	private ExecutorService executor = Executors.newFixedThreadPool(ths.length);
	private static final String insertSql =  "INSERT INTO t_distribution_log(id,partition_month,create_time,file_name,source_type,source_address,source_path,target_type,target_address,target_path,begin_time,end_time,result)"+
			 " VALUES  (seq_t_distribution_log.nextval,to_char(sysdate,'yyyy-mm'),sysdate,?,?,?,?,?,?,?,?,?,?)";
	public StoreDistributeProccessor(){
		for(int i =0;i<ths.length;i++){
			ths[i] = new StoreThread();
		}
	}
	private class StoreThread extends BaseProcessor{
		private Connection connection;
		StoreThread(){
			try {
				connection = DBHelper.getConnection();
				connection.setAutoCommit(false);
			} catch (SQLException e) {
				this.finish();
				logger.error("GetConnection Error", e);
				logger.error("StoreThread will not real running ....");
//				finish();
				e.printStackTrace();
			}
			
		}
		@Override
		public void run() {
			logger.info("Start StoreReportProccessor : " + this.getId());
			CallableStatement insertPs = null;
			ResultSet rs = null;
			try {
				insertPs = connection.prepareCall("{call insert_distribution(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)}");//connection.prepareStatement(insertSql);
				insertPs.registerOutParameter(15, Types.INTEGER);
			} catch (SQLException e) {
				e.printStackTrace();
				logger.error("分发入库线程初始化错误",e);
				this.finish();
			}
			while (!isFinish()) {
				int count = 0;
				while (!DistrbuteFileLogQueue.isEmpty() && count < BATCH_SIZE) {
					try {
						DistributionLog log = DistrbuteFileLogQueue.take();
						if (log == null) {
							continue;
						}
						// 写入分发日志
						logger.error("插入分发报文 " + log.getFile_name());
						//create_time
						//file_name
						insertPs.setString(1,log.getFile_name());
						//source_type
						insertPs.setString(2,log.getSource_type());
						//source_address
						insertPs.setString(3,log.getSource_address());
						//source_path
						insertPs.setString(4,log.getSource_path());
						//target_type
						insertPs.setString(5,log.getTarget_type());
						//target_address
						insertPs.setString(6,log.getTartget_address());
						//target_path
						insertPs.setString(7,log.getTarget_path());
						//begin_time
						insertPs.setTimestamp(8,new java.sql.Timestamp(log.getBegin_time().getTime()));
						//end_time
						insertPs.setTimestamp(9,new java.sql.Timestamp(log.getEnd_time().getTime()));
						//result
						insertPs.setString(10,log.getResult());
						insertPs.setString(11,log.getTarget_id());
						insertPs.setString(12,log.getStation_id());
						insertPs.setString(13,log.getFile_date());
						insertPs.setString(14,log.getFile_type());
						insertPs.setInt(15,0);
						insertPs.execute();
						connection.commit();
						
//						insertPs.addBatch();
					} catch (Exception e) {
						e.printStackTrace();
						logger.error("插入分发日志错误",e);
					}
					count++;
				}
				if (count == 0) {
					continue;
				}
				try {
					logger.info("提交插入分发");
//					insertPs.executeBatch();
					connection.commit();
//					insertPs.clearBatch();
				} catch (SQLException e) {
					logger.error("插入分发日志错误", e);
					e.printStackTrace();
					try {
						connection.rollback();
					} catch (SQLException e1) {
						e1.printStackTrace();
					}
				}
				if (DistrbuteFileLogQueue.isEmpty()) {
					// finish();
					// StoreReportProccessor.this.finish();
					// logger.info("结束入库线程");
				}
			}
		}
		
	}
	@Override
	public void run() {
		logger.info("开始分发入库线程");
		for(StoreThread thread:ths){
			executor.submit(thread);
		}
	}
	
}
